package gbench.sandbox.zk.junit;

import static gbench.common.matlib.MatlibCanvas.println;
import static gbench.common.tree.LittleTree.Term.FT;
import static gbench.common.tree.LittleTree.IRecord.REC;
import static gbench.common.tree.LittleTree.IRecord.A;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.Arrays;
import java.util.Optional;
import java.util.stream.*;

import org.apache.curator.framework.CuratorFramework;
import org.junit.jupiter.api.Test;

import gbench.common.matlib.data.DataReader.DFrame;
import gbench.commonApp.data.DataApp;
import gbench.sandbox.zk.zoo.IZKSession;
import gbench.sandbox.zk.zoo.ZKApp;
import gbench.sandbox.zk.zoo.ZKTrieNode;

/**
 * 
 * @author Administrator
 *
 */
public class JunitZk2 {

    @Test
    public void foo() {

        zkApp.withTransaction(sess -> {
            final var root = "/proj"; // 项目根目录
            sess.deleteIfExists(root);

            final var volume = dataApp //
                    .jdbcApply(jdbc -> jdbc.sql2maybe("select count(*) cnt from v_double_entry;")) // sql 语句
                    .map(e -> e.i4("cnt")).orElse(0); // 统计任务量
            final var pageSize = 1000; // 子任务的数据量大小
            final var times = (int) Math.ceil(volume * 1.0 / pageSize); // 子任务的个数
            final var latch = new CountDownLatch(times); // 子任务的同步控制

            Stream.iterate(0, i -> i < times, i -> i + 1).limit(times).forEach(i -> {
                final var start = pageSize * i; // 开始位置 包含
                final var end = Optional.of(pageSize * (i + 1)).map(j -> j > volume ? volume : j).get(); // 结束位置 不包含
                final var task_name = FT("task$0", i); // 节点名
                final var task_path = FT("$0/$1", root, task_name); // 节点路径
                final var range = A(start, end); // 任务数据范围
                final var task_descriptor = REC("taskid", i, "name", task_name, "path", task_path, "range", range); // 任务运行参数
                sess.ephemeralOf(task_path, task_descriptor); // 任务描述
            }); // OF

            final var pool = Executors.newFixedThreadPool(10); // 线程池
            sess.get_tree_stream_throws(root) // 提取节点
                    .filter(e -> e.isLeaf()) // 只提取叶子节点，叶子节点代表任务
                    .forEach(e -> { // 提取叶子节点
                        pool.execute(newTask(latch, sess, e)); // 执行任务
                    }); // forEach

            latch.await(); // 等待程序执行结束

            final var dfm = sess.recordOf(root).kvstream().collect(DFrame.dfmclc); // 生成结果矩阵
            println(dfm);
            println("total", dfm.colN(1).sum()); // 数据汇总
            sess.get_tree_throws(root).forEach(e -> { // 任务明细遍历
                final var line = FT("$0$1 ==> $2", " | ".repeat(e.getLevel() - 1), e.getName(), e.attrs());
                println(line);
            }); // forEach
        }); // withTransaction
    }

    /**
     * 创建任务子任务
     * 
     * @param latch 任务锁 本地同步
     * @param sess  会话对象 本地
     * @param task  子任务描述符
     * @return
     */
    public Runnable newTask(final CountDownLatch latch, IZKSession<CuratorFramework> sess, ZKTrieNode<String> task) {
        return () -> { // 创建任务
            final var name = task.getName(); // 任务名
            final var path = FT("$0", task.getPath()); // 子任务的路径
            final var descriptor = task.attrs(); // 提取子任务的属性
            final var range = descriptor.lla("range", Integer.class).toArray(Integer[]::new); // 数据范围
            final var sql = FT("select sum(total) sum,count(*) cnt from (select * from v_double_entry limit $0,$1) t",
                    range[0], range[1] - range[0]); // 数据计算语句
            println(sql);
            final var sum = dataApp.jdbcApply(jdbc -> jdbc.sql2maybe(sql)).map(r -> r.i4("sum")).orElse(0); // 数据计算
            final var tid = Thread.currentThread().getId(); // 当前线程的id
            sess.update(path, "sum", sum, "id", tid);// 提交任务语句
            sess.update(IZKSession.getParent(path), FT("$0$1", name, Arrays.asList(range)), sum); // 结论写入父节点
            latch.countDown(); // 任务完成
        }; // 创建任务
    }

    final String jdbcurl = "jdbc:mysql://localhost:3306/db?useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=True";
    final DataApp dataApp = new DataApp() { // mysql jdbc
    }.reload2(REC("url", jdbcurl));// 加载数据库配置
    final ZKApp zkApp = new ZKApp("localhost:2181,localhost:2182,localhost:2183");
}
